package ItemBased;

import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
//import org.apache.hadoop.mapreduce.Job;
//import org.apache.hadoop.mapreduce.Mapper;
//import org.apache.hadoop.mapreduce.Reducer;
//import org.apache.hadoop.mapreduce.Mapper.Context;
//import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
//import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import ItemBased.ToItemVectors.ToItemVectorsMap;
import ItemBased.ToItemVectors.ToItemVectorsReduce;

public class CollaborativeFiltering {
	
	public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
		 
	    JobConf job = new JobConf(CollaborativeFiltering.class);
	    job.setJarByClass(CollaborativeFiltering.class);
	    
	    job.setMapperClass(ToItemVectorsMap.class);
	    job.setReducerClass(ToItemVectorsReduce.class);	     

	    job.setInputFormat(TextInputFormat.class);
	    job.setOutputFormat(TextOutputFormat.class);
	    
	    job.setMapOutputKeyClass(LongWritable.class);
	    job.setMapOutputValueClass(LongPairWritableComparable.class);
	    
	    // set output key type   
	    job.setOutputKeyClass(LongWritable.class);
	    // set output value type
	    job.setOutputValueClass(LongPairArrayWritableComparable.class);
	    //set the HDFS path of the input data
	    FileInputFormat.setInputPaths(job, new Path(args[1]));
	    // set the HDFS path for the output
	    FileOutputFormat.setOutputPath(job, new Path(args[2]));
	    
	    JobClient.runJob(job);
	  }
}
